package com.rrd.dw.mr.gzip.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ucredit.babel.common.model.BaseModel;
import com.ucredit.babel.common.model.RecordType;
import com.ucredit.babel.common.model.third.baihang.BaihangQ1Data;
import com.ucredit.babel.common.model.third.baihang.BaihangQ1Record;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper.Context;

import java.io.IOException;

/*
 * 百行征信
 *
 * @author:szw
 * */

public class BaihangQ1RecordParse extends ParseBase{
    private Text val = new Text();
    private Text key = new Text();

    /**
     * @param context
     * @param model
     * @param successful
     * @return
     * @throws IOException
     * @throws InterruptedException
     */

    @Override
    public boolean parseJob(Context context, BaseModel<?> model, String successful) throws IOException, InterruptedException {
        BaihangQ1Record brRecord = (BaihangQ1Record) model;
        BaihangQ1Data brData = brRecord.getData();
        JSONObject data = JSON.parseObject(brData.getData());

        StringBuilder sb = getBaseInfo(brRecord, successful);
        StringBuilder bhSb = new StringBuilder(2000);

        JSONObject revolvingLoan = data.getJSONObject("revolvingLoan");
        JSONObject revolvingLoanSummary = revolvingLoan.getJSONObject("summary");
        JSONObject revolvingLoanD30 = revolvingLoan.getJSONObject("D30");
        JSONObject revolvingLoanD90 = revolvingLoan.getJSONObject("D90");
        JSONObject revolvingLoanD180 = revolvingLoan.getJSONObject("D180");
        JSONObject revolvingLoanD360 = revolvingLoan.getJSONObject("D360");

//        JSONObject financeLease = data.getJSONObject("financeLease");
        JSONObject nonRevolvingLoan = data.getJSONObject("nonRevolvingLoan");
        JSONObject nonRevolvingLoanSummary = nonRevolvingLoan.getJSONObject("summary");
        JSONObject nonRevolvingLoanD30 = nonRevolvingLoan.getJSONObject("D30");
        JSONObject nonRevolvingLoanD90 = nonRevolvingLoan.getJSONObject("D90");
        JSONObject nonRevolvingLoanD180 = nonRevolvingLoan.getJSONObject("D180");
        JSONObject nonRevolvingLoanD360 = nonRevolvingLoan.getJSONObject("D360");

//        JSONObject objectionLabeling = data.getJSONObject("objectionLabeling");
        JSONObject reportHeader = data.getJSONObject("reportHeader");
        JSONObject personalProfile = data.getJSONObject("personalProfile");
//        JSONObject personalStatement = data.getJSONObject("personalStatement");

        bhSb.append(revolvingLoanSummary.getString("remainingAmount")).append(CRT001);
        bhSb.append(revolvingLoanSummary.getString("remainingOverdueAmount")).append(CRT001);
        bhSb.append(revolvingLoanSummary.getString("revolvingLastCompensationDate")).append(CRT001);
        bhSb.append(revolvingLoanSummary.getString("accountCount")).append(CRT001);
        bhSb.append(revolvingLoanSummary.getString("remainingMaxOverdueStatus")).append(CRT001);
        bhSb.append(revolvingLoanSummary.getString("creditLimitSum")).append(CRT001);
        bhSb.append(revolvingLoanSummary.getString("validAccountCount")).append(CRT001);
        bhSb.append(revolvingLoanSummary.getString("remainingOverdueAccountCount")).append(CRT001);
        bhSb.append(revolvingLoanSummary.getString("maxOverdueStatus")).append(CRT001);
        bhSb.append(revolvingLoanSummary.getString("maxCreditLimitPerTenant")).append(CRT001);
        bhSb.append(revolvingLoanSummary.getString("overdueCount")).append(CRT001);

        bhSb.append(revolvingLoanD30.getString("accountCount")).append(CRT001);
        bhSb.append(revolvingLoanD30.getString("revolvingCompensationCount")).append(CRT001);
        bhSb.append(revolvingLoanD30.getString("revolvingCompensationAmount")).append(CRT001);
        bhSb.append(revolvingLoanD30.getString("creditLimitSum")).append(CRT001);
        bhSb.append(revolvingLoanD30.getString("revolvingCompensationTimes")).append(CRT001);
        bhSb.append(revolvingLoanD30.getString("overdueAccountCount")).append(CRT001);
        bhSb.append(revolvingLoanD30.getString("applyTenantCount")).append(CRT001);
        bhSb.append(revolvingLoanD30.getString("lendingAmount")).append(CRT001);

        bhSb.append(revolvingLoanD90.getString("accountCount")).append(CRT001);
        bhSb.append(revolvingLoanD90.getString("revolvingCompensationCount")).append(CRT001);
        bhSb.append(revolvingLoanD90.getString("revolvingCompensationAmount")).append(CRT001);
        bhSb.append(revolvingLoanD90.getString("creditLimitSum")).append(CRT001);
        bhSb.append(revolvingLoanD90.getString("revolvingCompensationTimes")).append(CRT001);
        bhSb.append(revolvingLoanD90.getString("overdueAccountCount")).append(CRT001);
        bhSb.append(revolvingLoanD90.getString("applyTenantCount")).append(CRT001);
        bhSb.append(revolvingLoanD90.getString("lendingAmount")).append(CRT001);

        bhSb.append(revolvingLoanD180.getString("accountCount")).append(CRT001);
        bhSb.append(revolvingLoanD180.getString("revolvingCompensationCount")).append(CRT001);
        bhSb.append(revolvingLoanD180.getString("revolvingCompensationAmount")).append(CRT001);
        bhSb.append(revolvingLoanD180.getString("creditLimitSum")).append(CRT001);
        bhSb.append(revolvingLoanD180.getString("revolvingCompensationTimes")).append(CRT001);
        bhSb.append(revolvingLoanD180.getString("overdueAccountCount")).append(CRT001);
        bhSb.append(revolvingLoanD180.getString("applyTenantCount")).append(CRT001);
        bhSb.append(revolvingLoanD180.getString("lendingAmount")).append(CRT001);

        bhSb.append(revolvingLoanD360.getString("accountCount")).append(CRT001);
        bhSb.append(revolvingLoanD360.getString("revolvingCompensationCount")).append(CRT001);
        bhSb.append(revolvingLoanD360.getString("revolvingCompensationAmount")).append(CRT001);
        bhSb.append(revolvingLoanD360.getString("creditLimitSum")).append(CRT001);
        bhSb.append(revolvingLoanD360.getString("revolvingCompensationTimes")).append(CRT001);
        bhSb.append(revolvingLoanD360.getString("overdueAccountCount")).append(CRT001);
        bhSb.append(revolvingLoanD360.getString("applyTenantCount")).append(CRT001);
        bhSb.append(revolvingLoanD360.getString("lendingAmount")).append(CRT001);

        bhSb.append(nonRevolvingLoanSummary.getString("lastCompensationDate")).append(CRT001);
        bhSb.append(nonRevolvingLoanSummary.getString("remainingAmount")).append(CRT001);
        bhSb.append(nonRevolvingLoanSummary.getString("remainingOverdueLoanCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanSummary.getString("remainingOverdueAmount")).append(CRT001);
        bhSb.append(nonRevolvingLoanSummary.getString("remainingMaxOverdueStatus")).append(CRT001);
        bhSb.append(nonRevolvingLoanSummary.getString("loanCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanSummary.getString("maxOverdueStatus")).append(CRT001);
        bhSb.append(nonRevolvingLoanSummary.getString("openLoanCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanSummary.getString("overdueCount")).append(CRT001);

        bhSb.append(nonRevolvingLoanD30.getString("loanTenantCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD30.getString("compensationCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD30.getString("averageLoanAmount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD30.getString("overdueLoanCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD30.getString("compensationAmount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD30.getString("loanCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD30.getString("maxLoanAmount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD30.getString("applyTenantCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD30.getString("loanAmount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD30.getString("compensationTimes")).append(CRT001);

        bhSb.append(nonRevolvingLoanD90.getString("loanTenantCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD90.getString("compensationCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD90.getString("averageLoanAmount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD90.getString("overdueLoanCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD90.getString("compensationAmount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD90.getString("loanCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD90.getString("maxLoanAmount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD90.getString("applyTenantCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD90.getString("loanAmount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD90.getString("compensationTimes")).append(CRT001);

        bhSb.append(nonRevolvingLoanD180.getString("loanTenantCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD180.getString("compensationCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD180.getString("averageLoanAmount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD180.getString("overdueLoanCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD180.getString("compensationAmount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD180.getString("loanCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD180.getString("maxLoanAmount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD180.getString("applyTenantCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD180.getString("loanAmount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD180.getString("compensationTimes")).append(CRT001);

        bhSb.append(nonRevolvingLoanD360.getString("loanTenantCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD360.getString("compensationCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD360.getString("averageLoanAmount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD360.getString("overdueLoanCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD360.getString("compensationAmount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD360.getString("loanCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD360.getString("maxLoanAmount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD360.getString("applyTenantCount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD360.getString("loanAmount")).append(CRT001);
        bhSb.append(nonRevolvingLoanD360.getString("compensationTimes")).append(CRT001);

        bhSb.append(data.getString("queryHistory")).append(CRT001);
        bhSb.append(reportHeader.getString("reportId")).append(CRT001);
        bhSb.append(reportHeader.getString("queryResult")).append(CRT001);
        bhSb.append(reportHeader.getString("reportTime")).append(CRT001);

        bhSb.append(personalProfile.getString("name")).append(CRT001);
        bhSb.append(personalProfile.getString("mobileCount")).append(CRT001);
        bhSb.append(personalProfile.getString("pid")).append(CRT001);

        bhSb.append(data.getString("homeInfo")).append(CRT001);
        bhSb.append(data.getString("workInfo")).append(CRT001);
        key.set(getKey());
        val.set((sb.toString() + bhSb.toString()).replace("\n", " ").replace("\r", " "));
        context.write(key, val);

        return true;
    }

    @Override
    public String getTableName() {
        return "kafka_babel_bh_q1_record_inc";
    }

    @Override
    public String getType() {
        return RecordType.BAIHANG_NEW_Q1_RECORD.toString();
    }

    @Override
    public String getComment() {
        return " 百行征信";
    }
}
